package com.atguigu.flink.chapter11;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


public class Flink07_SQL_File {
       public static void main(String[] args) throws Exception {
               Configuration configuration = new Configuration();
               //web  UI端口
               configuration.setInteger("rest.prot",10000);
           StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
               env.setParallelism(1);
           StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

           //建立一个表，与文件关联
            tableEnv
                   .executeSql(
                           "create table sensor(" +
                                   "  id string," +
                                   " ts bigint," +
                                   " vc int " +
                                   ")with (" +
                                   "   'connector'='filesystem'," +
                                   "    'path' = 'input/sensor.txt'," +
                                   "    'format'= 'csv' )");

           Table table = tableEnv.sqlQuery("select * from sensor");

           //再建立一个表，与sink  输出文件 关联
            tableEnv
                   .executeSql(
                           "create table out1(" +
                                   "  id string," +
                                   " ts bigint," +
                                   " vc int " +
                                   ")with (" +
                                   "   'connector'='filesystem'," +
                                   "    'path' = 'input/aa.txt'," +
                                   "    'format'= 'csv' )");
            // sql 里 out是特殊字符

            table.executeInsert("out1");







       }
}
